package com.tipdm.scalaDemo

import org.apache.log4j.{Level, Logger}
import org.apache.spark.sql.SparkSession
import org.apache.spark.sql.functions._
import org.apache.spark.{SparkConf, SparkContext}

object t2 {
  def main(args: Array[String]): Unit = {
    // 初始化 Spark 环境
    val conf = new SparkConf().setAppName("wordcount").setMaster("local")
    val sc = new SparkContext(conf)
    val spark = SparkSession.builder().appName("test").master("local").getOrCreate()
    Logger.getLogger("org").setLevel(Level.OFF)
    import spark.implicits._

    // 读取订单主表
    val nwDF = spark.read.option("header", "true").option("inferSchema", "true")
      .csv("D:\\迅雷下载\\1718240841504010398\\NW-Orders-01.csv")
    // 字段示例: OrderID, CustomerID, EmployeeID, OrderDate, ShipCountry

    // 读取订单明细表
    val nwDF2 = spark.read.option("header", "true").option("inferSchema", "true")
      .csv("D:\\迅雷下载\\1718240854877025993\\NW-Order-Details.csv")
    // 字段示例: OrderID, ProductID, UnitPrice, Qty, Discount

    // 每个客户下了多少订单
    nwDF.groupBy("CustomerID").count().show()

    // 每个国家的订单数量
    nwDF.groupBy("ShipCountry").count().show()

    // 连接订单主表和明细表
    val joinDF = nwDF.join(nwDF2, Seq("OrderID"), "inner")

    // 提取订单日期中的年份和月份
    val addDateDF = joinDF
      .withColumn("year", year(col("OrderDate")))
      .withColumn("month", month(col("OrderDate")))

    // 每年订单数量
    addDateDF.groupBy("year").count().show()

    // 每月订单数量
    addDateDF.groupBy("month").count().show()

    // 计算每行订单的销售额：UnitPrice * Qty * (1 - Discount)
    val ordersWithTotalPrice = addDateDF.withColumn("TotalPrice", expr("UnitPrice * Qty * (1 - Discount)"))

    // 每个客户每年的销售总额
    val customerSalesPerYear = ordersWithTotalPrice
      .groupBy("CustomerID", "year")
      .agg(sum("TotalPrice").alias("TotalSales"))
    customerSalesPerYear.show()

    // 每个客户每年的平均订单数量（推荐做法）
    val customerYearOrderCount = ordersWithTotalPrice
      .groupBy("CustomerID", "year")
      .agg(count("*").alias("yearly_order_count"))

    val avgOrdersPerYear = customerYearOrderCount
      .groupBy("CustomerID")
      .agg(avg("yearly_order_count").alias("AvgOrdersPerYear"))

    avgOrdersPerYear.show()
  }
}
